//@Auth:zdl
package nsq

import (
	"encoding/json"
	"gitee.com/sky_big/workmod/myLog"
	"github.com/gogf/gf/g"
	"github.com/nsqio/go-nsq"
	"log"
)

var (
	topic        string
	addr         string
	receiveTopic string
	channel      string
	producer *nsq.Producer
)

type Handler func(data []byte)error

func Init(){
	topic = g.Config().GetString("nsq.topic")
	addr = g.Config().GetString("nsq.addr")
	channel=g.Config().GetString("nsq.receive.channel")
	receiveTopic = g.Config().GetString("nsq.receive.topic")
	//初始化推送
	cfg := nsq.NewConfig()
	var  err error
	producer, err = nsq.NewProducer(addr, cfg)
	if err != nil {
		log.Fatal(err)
	}
}

func PublishStart(data interface{},topics... string) {
	if producer==nil||producer.Ping()!=nil{
		Init()
	}
	tpc:=topic
	if len(topics)>0{
		tpc=topics[0]
	}
	bytes,err:=json.Marshal(data)
	if err!=nil{
		myLog.Error(err)
	}else{
		if err := producer.Publish(tpc,bytes ); err != nil {//[]byte("test message")   "NBMobile_receive"
			log.Fatal("publish error: " + err.Error())
		}
	}
	//time.Sleep(1 * time.Second)
}

// 消费者
func ConsumerStart(handle Handler,topics...string) {
	tpc:=topic
	if len(topics)>0{
		tpc=topics[0]
	}
	cfg := nsq.NewConfig()
	consumer, err := nsq.NewConsumer(tpc, channel, cfg)
	if err != nil {
		log.Fatal(err)
	}
	// 设置消息处理函数
	consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
		log.Println(message.Body)
		err:=handle(message.Body)
		return err
	}))
	// 连接到单例nsqd
	if err := consumer.ConnectToNSQD(addr); err != nil {
		log.Fatal(err)
	}
	<-consumer.StopChan
}
